package com.central.scheduler.worker.runner;

import com.central.scheduler.service.bean.SpringApplicationContext;
import com.central.scheduler.service.queue.entity.TaskExecutionContext;
import com.central.scheduler.common.enums.Event;
import com.central.scheduler.common.enums.ExecutionStatus;
import com.central.scheduler.common.thread.Stopper;
import com.central.scheduler.common.thread.ThreadUtils;
import com.central.scheduler.common.utils.JSONUtils;
import com.central.scheduler.remote.command.TaskExecuteResponseCommand;
import com.central.scheduler.worker.cache.ResponceCache;
import com.central.scheduler.worker.config.WorkerConfig;
import com.central.scheduler.worker.processor.TaskCallbackService;
import com.central.scheduler.spi.task.TaskExecutionContextCacheManager;
import com.central.scheduler.spi.task.request.TaskRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author Tindy
 * @date 2021/10/29
 * @describe Manage tasks
 */
@Component
public class WorkerManagerThread implements Runnable {
    private final Logger logger = LoggerFactory.getLogger(WorkerManagerThread.class);
    /**
     * task queue
     */
    private final DelayQueue<TaskExecuteThread> workerExecuteQueue = new DelayQueue<>();
    /**
     * thread executor service
     */
    private final ExecutorService workerExecService;

    /**
     * worker config
     */
    private WorkerConfig workerConfig;

    /**
     * task callback service
     */
    @Autowired
    private TaskCallbackService taskCallbackService;


    /**
     * get thread pool queue size
     *
     * @return queue size
     */
    public int getThreadPoolQueueSize() {
        return ((ThreadPoolExecutor) workerExecService).getQueue().size();
    }
    public WorkerManagerThread() {
        this.workerConfig = SpringApplicationContext.getBean(WorkerConfig.class);
        this.workerExecService = ThreadUtils.newDaemonFixedThreadExecutor("Worker-Execute-Thread", this.workerConfig.getWorkerExecThreads());
    }

    public void start() {
        Thread thread = new Thread(this, this.getClass().getName());
        thread.setDaemon(true);
        thread.start();
    }

    @Override
    public void run() {
        Thread.currentThread().setName("Worker-Execute-Manager-Thread");
        TaskExecuteThread taskExecuteThread;
        while (Stopper.isRunning()) {
            try {
                taskExecuteThread = workerExecuteQueue.take();
                workerExecService.submit(taskExecuteThread);
            } catch (Exception e) {
                logger.error("An unexpected interrupt is happened, "
                        + "the exception will be ignored and this thread will continue to run", e);
            }
        }
    }
    /**
     * Kill tasks that have not been executed, like delay task
     * then send Response to Master, update the execution status of task instance
     */
    public void killTaskBeforeExecuteByInstanceId(Long taskInstanceId) {
        workerExecuteQueue.stream()
                .filter(taskExecuteThread -> taskExecuteThread.getTaskExecutionContext().getTaskInstanceId().equals(taskInstanceId))
                .forEach(workerExecuteQueue::remove);
        sendTaskKillResponse(taskInstanceId);
    }
    /**
     * kill task before execute , like delay task
     */
    private void sendTaskKillResponse(Long taskInstanceId) {
        TaskRequest taskRequest = TaskExecutionContextCacheManager.getByTaskInstanceId(taskInstanceId);
        if (taskRequest == null) {
            return;
        }
        TaskExecutionContext taskExecutionContext = JSONUtils.parseObject(JSONUtils.toJsonString(taskRequest), TaskExecutionContext.class);
        TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand(taskExecutionContext.getTaskInstanceId(), taskExecutionContext.getProcessInstanceId());
        responseCommand.setStatus(ExecutionStatus.KILL.getCode());
        ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
        taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
    }
    /**
     * submit task
     *
     * @param taskExecuteThread taskExecuteThread
     * @return submit result
     */
    public boolean offer(TaskExecuteThread taskExecuteThread) {
        return workerExecuteQueue.offer(taskExecuteThread);
    }
    /**
     * get delay queue size
     *
     * @return queue size
     */
    public int getDelayQueueSize() {
        return workerExecuteQueue.size();
    }
}
